package com.ddone.demo;

import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;

/**
 * @author ddone
 * @date 2024/5/27-00:34
 */
public class ConfigReaderDemo {
    public static void main(String[] args) {
        // 加载配置文件
        Config config = ConfigFactory.load("config.conf");

        // 读取 Flink 配置
        Config flinkConfig = config.getConfig("application.flink");
        int parallelism = flinkConfig.getInt("parallelism");
        int lookupParallelism = flinkConfig.getInt("lookup.uds.parallelism");
        String checkpointPath = flinkConfig.getString("checkpoint.path");
        int checkpointInterval = flinkConfig.getInt("checkpoint.interval");
        int asyncTimeout = flinkConfig.getInt("async.timeout.seconds");
        int asyncCapacity = flinkConfig.getInt("async.capacity");

        System.out.println("Flink Configuration:");
        System.out.println("Parallelism: " + parallelism);
        System.out.println("Lookup UDS Parallelism: " + lookupParallelism);
        System.out.println("Checkpoint Path: " + checkpointPath);
        System.out.println("Checkpoint Interval: " + checkpointInterval);
        System.out.println("Async Timeout: " + asyncTimeout);
        System.out.println("Async Capacity: " + asyncCapacity);

        // 读取 Application 配置
        Config appConfig = config.getConfig("application.application");
        String consumerGroupId = appConfig.getString("consumer.group.id");
        String offsetAutoReset = appConfig.getString("offset.auto.reset");

        System.out.println("\nApplication Configuration:");
        System.out.println("Consumer Group ID: " + consumerGroupId);
        System.out.println("Offset Auto Reset: " + offsetAutoReset);

        // 读取 Kafka 配置
        Config kafkaConfig = config.getConfig("application.kafka");

        // Datasight Kafka brokers
        Config datasightConfig = kafkaConfig.getConfig("datasight");
        String datasightBrokers = datasightConfig.getString("brokers");
        String datasightUsername = datasightConfig.getString("username");
        String datasightPassword = datasightConfig.getString("password");

        System.out.println("\nDatasight Kafka Configuration:");
        System.out.println("Brokers: " + datasightBrokers);
        System.out.println("Username: " + datasightUsername);
        System.out.println("Password: " + datasightPassword);

        // Common Kafka brokers
        Config commonConfig = kafkaConfig.getConfig("common");
        String commonBrokers = commonConfig.getString("brokers");
        String commonUsername = commonConfig.getString("username");
        String commonPassword = commonConfig.getString("password");

        System.out.println("\nCommon Kafka Configuration:");
        System.out.println("Brokers: " + commonBrokers);
        System.out.println("Username: " + commonUsername);
        System.out.println("Password: " + commonPassword);

        // Kafka source and sink topics
        String sourceTopic = kafkaConfig.getConfig("source.thesis").getString("topic.name.campaign");
        String sinkTopic = kafkaConfig.getConfig("sink.campaign").getString("topic.name");

        System.out.println("\nKafka Topics:");
        System.out.println("Source Topic: " + sourceTopic);
        System.out.println("Sink Topic: " + sinkTopic);

        // 读取 MySQL 配置
        Config mysqlConfig = config.getConfig("application.mysql.thesis");
        String mysqlUrl = mysqlConfig.getString("url");
        String mysqlDatabase = mysqlConfig.getString("database");
        String mysqlUsername = mysqlConfig.getString("username");
        String mysqlPassword = mysqlConfig.getString("password");

        System.out.println("\nMySQL Configuration:");
        System.out.println("URL: " + mysqlUrl);
        System.out.println("Database: " + mysqlDatabase);
        System.out.println("Username: " + mysqlUsername);
        System.out.println("Password: " + mysqlPassword);

        // 读取 Cache 配置
        Config cacheConfig = config.getConfig("application.cache");
        int cacheExpireSeconds = cacheConfig.getInt("cache_write_expire_seconds");
        int cacheMaxSize = cacheConfig.getInt("cache_max_size");

        System.out.println("\nCache Configuration:");
        System.out.println("Cache Write Expire Seconds: " + cacheExpireSeconds);
        System.out.println("Cache Max Size: " + cacheMaxSize);
    }
}
